package com.qf.smsplatform.interfaces.mq;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.qf.smsplatform.common.constants.CacheConstants;
import com.qf.smsplatform.common.constants.RabbitMQConstants;
import com.qf.smsplatform.common.model.StandardReport;
import com.qf.smsplatform.interfaces.config.DelayedConfig;
import com.qf.smsplatform.interfaces.service.api.CacheService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;


import java.io.IOException;
import java.util.Map;

/**
 * @Author: LeeRui
 * 2021/2/1 下午 7:20
 **/
@Component
@Slf4j
public class ReportTopicMQ {

    @Autowired
    private CacheService cacheService;

    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private Integer[] sendTTL = {0, 10000, 30000, 60000, 300000};

    //第一次接收MQ消息进行消费并推送报告
    @RabbitListener(queues = {RabbitMQConstants.TOPIC_PUSH_SMS_REPORT})
    public void reportTopicMQ(StandardReport report, Channel channel, Message message) throws IOException {
        //获取当前客户的id用于后续客户信息查询及操作
        long clientID = report.getClientID();
        //缓存模块查询当前客户的信息
        Map map = cacheService.hMGet(CacheConstants.CACHE_PREFIX_CLIENT + clientID);
        //判断当前客户的isreturnstatus是否为1 （String 不能被转换为 Integer）
        Integer isreturnstatus = Integer.parseInt((String) map.get("isreturnstatus"));
        //判断用户是否要求返回推送报告
        if (isreturnstatus != null && isreturnstatus == 1) {
            //查询当前客户的接收状态报告的http路径
            String receivestatusurl = (String) map.get("receivestatusurl");
            //将客户推送报告请求url封装到report中
            report.setReceiveStatusUrl(receivestatusurl);
            //记录发送次数，第一次发送
            report.setSendCount(1);
            String result = null;
            try {
                //封装请求参数并通过restTemplate发送请求到http路径
                //封装HttpHeaders指定编码
                HttpHeaders httpHeaders = new HttpHeaders();
                httpHeaders.setContentType(MediaType.APPLICATION_JSON_UTF8);
                //封装HttpEntity
                HttpEntity httpEntity = new HttpEntity(objectMapper.writeValueAsString(report), httpHeaders);
                //发送请求
                result = restTemplate.postForObject(report.getReceiveStatusUrl(), httpEntity, String.class);
                //出现了异常 -- 记录日志，并发送消息到延迟队列（只重发4次）
            } catch (JsonProcessingException | RestClientException e) {
                if (report.getSendCount() != 5) {
                    log.info("【接口模块】 第{}次用户报告推送失败！！！重新发送中。。。", report.getSendCount());
                    rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "routingkey", report, new MessagePostProcessor() {
                        @Override
                        public Message postProcessMessage(Message message) throws AmqpException {
                            //根据推送次数到sendTTL里取值作为本次延迟时间
                            message.getMessageProperties().setDelay(sendTTL[report.getSendCount()]);
                            System.out.println("本次延迟推送时间为："+(sendTTL[report.getSendCount()])/1000+"秒!!!重新尝试推送次数还剩："+(4-report.getSendCount())+"次");
                            return message;
                        }
                    });
                    return;
                }
            }
            //正常返回，结果不是SUCCESS -- 记录日志，并发送消息到延迟队列（只重发4次）
            if (!"SUCCESS".equals(result)) {
                if (report.getSendCount() != 5) {
                    log.info("【接口模块】 第{}次用户报告推送失败！！！重新发送中。。。", report.getSendCount());
                    //手动ACK
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "routingkey", report, new MessagePostProcessor() {
                        @Override
                        public Message postProcessMessage(Message message) throws AmqpException {
                            //根据推送次数到sendTTL里取值作为本次延迟时间
                            message.getMessageProperties().setDelay(sendTTL[report.getSendCount()]);
                            System.out.println("本次延迟推送时间为："+(sendTTL[report.getSendCount()])/1000+"秒!!!重新尝试推送次数还剩："+(4-report.getSendCount())+"次");
                            return message;
                        }
                    });
                }
                return;
            }
            //正常返回，结果是SUCCESS -- 记录日志，ack并且return
            log.info("【接口模块】 用户报告推送成功！！！");
            //手动ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        } else {
            log.info("【接口模块】 用户要求无需返回推送报告！！！ isreturnstatus={} ", isreturnstatus);
            //手动ACK
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }

    }

    //首次推送报告失败，MQ自产自销式再尝试给客户推送几次报告
    @RabbitListener(queues = {DelayedConfig.DELAYED_QUEUE})
    public void selfConsumer(StandardReport report, Channel channel, Message message) throws IOException {
        //重新封装report中的推送次数
        report.setSendCount(report.getSendCount() + 1);
        //判断重新推送次数
        String result = null;
        try {
            //封装请求再次使用延时队列推送报告
            //封装HttpHeaders指定编码
            HttpHeaders httpHeaders = new HttpHeaders();
            httpHeaders.setContentType(MediaType.APPLICATION_JSON_UTF8);
            //封装httpEntity
            HttpEntity httpEntity = new HttpEntity(objectMapper.writeValueAsString(report), httpHeaders);
            //发送请求
            result = restTemplate.postForObject(report.getReceiveStatusUrl(), httpEntity, String.class);
            //出现了异常 -- 记录日志，并发送消息到延迟队列（只重发4次），ack并且return（不要忘记return）
        } catch (JsonProcessingException | RestClientException e) {
            if (report.getSendCount() != 5) {
                log.info("【接口模块】 第{}次用户报告推送失败！！！重新发送中。。。", report.getSendCount());
                rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "routingkey", report, new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        //根据推送次数到sendTTL里取值作为本次延迟时间
                        message.getMessageProperties().setDelay(sendTTL[report.getSendCount()]);
                        System.out.println("本次延迟推送时间为："+(sendTTL[report.getSendCount()])/1000+"秒!!!重新尝试推送次数还剩："+(4-report.getSendCount())+"次");
                        return message;
                    }
                });
            }
            return;
        }
        //若推送报告失败，则记录日志，并将消息发送到延迟队列中，然后手动ack
        if (!"SUCCESS".equals(result)) {
            if (report.getSendCount() != 5) {
                log.info("【接口模块】 第{}次用户报告推送失败！！！重新发送中。。。", report.getSendCount());
                rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "routingkey", report, new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        //根据推送次数到sendTTL里取值作为本次延迟时间
                        message.getMessageProperties().setDelay(sendTTL[report.getSendCount()]);
                        System.out.println("本次延迟推送时间为："+(sendTTL[report.getSendCount()])/1000+"秒!!!重新尝试推送次数还剩："+(4-report.getSendCount())+"次");
                        return message;
                    }
                });
            }
            return;
        }
        //正常返回，结果是SUCCESS -- 记录日志，ack并且return
        log.info("【接口模块】 用户报告推送成功！！！");
        //手动ack
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        return;
    }
}
